import asyncio
import random
import numpy as np
import time
import torch
import torchaudio
from config import USER_VOICE_FOLDER, max_val, SECRET_KEY, ALGORITHM

import os
import librosa
from loguru import logger
import uuid
from jose import jwt, JWTError
import httpx
from pydub import AudioSegment
from server.db.repository import get_voice_clone_task_by_timbre_id
from fastapi import HTTPException, Request



# 验证音频文件时长是否在限制范围内
def is_audio_duration_within_limit(file_path, limit_seconds=30):
    try:
        # 获取音频文件的信息
        audio_info = torchaudio.info(file_path)

        # 计算音频时长
        duration_seconds = audio_info.num_frames / audio_info.sample_rate

        # 判断音频时长是否在限制范围内
        if duration_seconds <= limit_seconds:
            logger.debug(f"音频时长为 {duration_seconds:.2f} 秒，在 {limit_seconds} 秒以内")
            return True
        else:
            logger.debug(f"音频时长为 {duration_seconds:.2f} 秒，超过了 {limit_seconds} 秒")
            return False

    except Exception as e:
        logger.error(f"无法处理音频文件 {file_path}: {e}")
        return False


# 后处理音频函数，进行音频裁剪、归一化并填充
def postprocess(speech, target_sr, top_db=60, hop_length=220, win_length=440):
    """
    后处理音频，去除静音部分并归一化音量。

    :param speech: 输入的音频数组
    :param target_sr: 目标采样率
    :param top_db: 静音阈值
    :param hop_length: 窗口步长
    :param win_length: 窗口大小
    :return: 处理后的音频
    """
    # 裁剪静音部分
    speech, _ = librosa.effects.trim(speech, top_db=top_db, frame_length=win_length, hop_length=hop_length)

    # 归一化处理，防止音量过大
    if speech.abs().max() > max_val:
        speech = speech / speech.abs().max() * max_val

    # 为音频末尾添加0.2秒的静音
    speech = torch.concat([speech, torch.zeros(1, int(target_sr * 0.2))], dim=1)

    return speech


# 获取demo音频路径
def get_demo_audio_path(uid, timbre_id, demo_audio_dir):
    os.makedirs(demo_audio_dir, exist_ok=True)
    return os.path.join(demo_audio_dir, f"{uid}_{timbre_id}_demo.wav")


# 获取用户音色文件
def get_timbre_file(uid, timbre_id):
    timbre_folder = os.path.join(USER_VOICE_FOLDER, uid, timbre_id)
    # 如果timbre_folder不存在，返回None
    if os.path.isdir(timbre_folder):
        timbre_file = os.listdir(timbre_folder)[0]
        timbre_file = os.path.join(timbre_folder, timbre_file) if timbre_file else None
    else:
        timbre_file = None

    return timbre_file


# 设置所有随机种子
def set_all_random_seed(seed):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)




def run_async(cor):
    '''
    在同步环境中运行异步代码
    '''
    try:
        loop = asyncio.get_running_loop()
    except:
        loop = asyncio.new_event_loop()
    return loop.run_until_complete(cor)

def generate_timbre_id() -> int:
    '''
    生成音色ID保证唯一性且控制为int32范围内
    '''
    timestamp = int(time.time())  # 当前时间戳（秒）
    uuid_int = uuid.uuid4().int & 0x7FFFFFFF  # 保证结果在 0 到 2^31-1 范围内
    result = (timestamp ^ uuid_int) & 0x7FFFFFFF  # 保证结果在 0 到 2^31-1 范围内
    return result

def decode_access_token(token: str) -> dict:
    """
    解码JWT Token

    Args:
        token (str): 需要解码的JWT Token

    Returns:
        dict: 解码后的Token数据

    Raises:
        JWTError: 如果Token无效或过期
    """
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError as e:
        raise JWTError(f"Token解码失败: {e}")


def verify_token(token: str, decode_field: str, decode_field_value: str) -> dict:
    '''
    验证Token是否有效
    '''
    try:
        token_detail = decode_access_token(token)
        logger.info(f"Token解码成功: {token_detail}")
        token_value = token_detail.get(decode_field)

        # 如果token_value为空，返回False
        if not token_value:
            logger.error(f"TokenValue值为空")
            return {"status": "-1", "message": "TokenValue值为空"}
        # 如果token_value与field_value相等，返回True
        if token_value == decode_field_value:
            return {"status": "0", "message": "Token验证成功"}
        else:
            logger.error(f"TokenValue: {token_value} 与 FieldValue: {decode_field_value} 不匹配")
            return {"status": "-2", "message": f"TokenValue: {token_value} 与 FieldValue: {decode_field_value} 不匹配"}
    except JWTError as e:
        logger.error(f"Token验证失败: {e}")
        return {"status": "-3", "message": f"Token验证失败: {e}"}

# 获取token
def get_token(request: Request):
    """
    获取token
    request: 请求对象
    """
    token = request.headers.get("Authorization")
    if not token:
        raise HTTPException(
            status_code=401,
            detail={"type": "token_missing", "msg": "未提供token"}
        )
    return token

def decode_verify_token(request: Request, secret_key: str, algorithm: str):
    """
    解码token
    request: 请求对象
    secret_key: JWT密钥
    algorithm: JWT算法
    """
    try:
        # 获取token
        token = get_token(request)
        # 解码token
        payload = jwt.decode(
            token, 
            secret_key, 
            algorithms=[algorithm],
            options={"verify_exp": True}
        )
        logger.info(f"token解码成功: {payload}")
        # 返回payload
        return payload
    except jwt.ExpiredSignatureError as e:
        logger.error(f"token已过期: {e}")
        raise HTTPException(
            status_code=401,
            detail={"type": "token_expired", "msg": "token已过期", "error": e}  # 使用结构化错误信息
        )
    except JWTError as e:
        logger.error(f"解析token失败: {e}")
        raise HTTPException(
            status_code=401,
            detail={"type": "token_invalid", "msg": "解析token失败", "error": e}  # 明确错误类型
        )
    

async def download_timbre_file(uid, timbre_id):
    try:
        result = get_voice_clone_task_by_timbre_id(timbre_id)
        if result:
            timbre_url = result["url"]
            user_directory = os.path.join(USER_VOICE_FOLDER, uid, timbre_id)
            os.makedirs(user_directory, exist_ok=True)
            object_name = os.path.basename(timbre_url)
            local_file_path = os.path.join(user_directory, object_name)
            async with httpx.AsyncClient() as client:
                response = await client.get(timbre_url)
                if response.status_code == 200:
                    with open(local_file_path, "wb") as f:
                        f.write(response.content)
                else:
                    logger.error(f"语音合成接口下载音色文件时出现错误: {response.text}")
                    return False, None
            return True, local_file_path
    except Exception as e:
        logger.error(f"获取音色文件时出现错误: {str(e)}")
        return False, None

def postprocess_timbre_file(local_file_path):
    try:
        audio = AudioSegment.from_file(local_file_path)

        audio.export(local_file_path, format="wav")
        return True
    except Exception as e:
        logger.error(f"音色文件格式转换时出现错误: {str(e)}")
        return False

